package com.ultrapower.mapper;

import com.sematext.hbase.wd.RowKeyDistributorByHashPrefix;
import com.ultrapower.bean.ApplicationBean;
import com.ultrapower.buffer.Buffer;
import com.ultrapower.buffer.FixedBuffer;
import com.ultrapower.util.ApplicationMapStatisticsUtils;
import com.ultrapower.util.CommonUtils;
import com.ultrapower.util.topology.LinkDataMap;
import com.ultrapower.util.topology.LinkFilter;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * @author wangy
 * 
 */
public class MapStatisticsCalleeMapper implements RowMapper<LinkDataMap> {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private final LinkFilter filter;


    private RowKeyDistributorByHashPrefix rowKeyDistributorByHashPrefix ;

    public MapStatisticsCalleeMapper(RowKeyDistributorByHashPrefix rowKeyDistributorByHashPrefix) {
        this.rowKeyDistributorByHashPrefix=rowKeyDistributorByHashPrefix;
        this.filter = LinkFilter.FILTER;
    }

    @Override
    public LinkDataMap mapRow(Result result, int rowNum) throws Exception {
        if (result.isEmpty()) {
            return new LinkDataMap();
        }
        logger.debug("mapRow:{}", rowNum);

        final byte[] rowKey = getOriginalKey(result.getRow());

        final Buffer row = new FixedBuffer(rowKey);
        final ApplicationBean calleeApplication = readCalleeApplication(row);
        final long timestamp = CommonUtils.recoveryTimeMillis(row.readLong());

        final LinkDataMap linkDataMap = new LinkDataMap();
        for (Cell cell : result.rawCells()) {

            final byte[] qualifier = CellUtil.cloneQualifier(cell);
            final ApplicationBean callerApplication = readCallerApplication(qualifier, calleeApplication.getServiceTypeCode());
            if (filter.filter(callerApplication)) {
                continue;
            }

            long requestCount = Bytes.toLong(cell.getValueArray(), cell.getValueOffset());
            short histogramSlot = ApplicationMapStatisticsUtils.getHistogramSlotFromColumnName(qualifier);

            String callerHost = ApplicationMapStatisticsUtils.getHost(qualifier);
            // There may be no callerHost for virtual queue nodes from user-defined entry points.
            // Terminal nodes, such as httpclient will not have callerHost set as well, but since they're terminal
            // nodes, they would not have reached here in the first place.
//            if (calleeApplication.getServiceType().isQueue()) {
//                callerHost = StringUtils.defaultString(callerHost);
//            }
            callerHost = callerHost==null?"":callerHost;
            boolean isError = histogramSlot == (short) -1;

            if (logger.isDebugEnabled()) {
                logger.debug("    Fetched Callee. {} callerHost:{} -> {} (slot:{}/{}),  ", callerApplication, callerHost, calleeApplication, histogramSlot, requestCount);
            }

            final short slotTime = (isError) ? (short) -1 : histogramSlot;
            linkDataMap.addLinkData(callerApplication, callerApplication.getName(), calleeApplication, callerHost, timestamp, slotTime, requestCount);

            if (logger.isDebugEnabled()) {
                logger.debug("    Fetched Callee. statistics:{}", linkDataMap);
            }
        }

        return linkDataMap;
    }

    private ApplicationBean readCallerApplication(byte[] qualifier, short calleeServiceType) {
        short callerServiceType = ApplicationMapStatisticsUtils.getDestServiceTypeFromColumnName(qualifier);
        // Caller may be a user node, and user nodes may call nodes with the same application name but different service type.
        // To distinguish between these user nodes, append callee's service type to the application name.
        String callerApplicationName;
        //ServiceType USER = of(2, "USER", RECORD_STATISTICS);
        if (callerServiceType == 2) {
            callerApplicationName = ApplicationMapStatisticsUtils.getDestApplicationNameFromColumnNameForUser(qualifier, calleeServiceType);
        } else {
            callerApplicationName = ApplicationMapStatisticsUtils.getDestApplicationNameFromColumnName(qualifier);
        }
        return  new ApplicationBean(callerApplicationName, calleeServiceType);
        //return this.applicationFactory.createApplication(callerApplicationName, callerServiceType);
    }

    private ApplicationBean readCalleeApplication(Buffer row) {
        String calleeApplicationName = row.read2PrefixedString();
        short calleeServiceType = row.readShort();
        return new ApplicationBean(calleeApplicationName, calleeServiceType);
        //return this.applicationFactory.createApplication(calleeApplicationName, calleeServiceType);
    }

    private byte[] getOriginalKey(byte[] rowKey) {
        return rowKeyDistributorByHashPrefix.getOriginalKey(rowKey);
    }
    
}
